package com.luodea.rabbitmqconsume;

import com.luodea.common.RabbitMqConfig;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpMessageHeaderAccessor;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.io.UnsupportedEncodingException;


/**
 * ackMode = "manual" 开启确认消费模式
 */
@RabbitListener(queues = RabbitMqConfig.Headers.Queue.IMAGE,ackMode = "MANUAL")
@Component
public class AckConsume {

    @RabbitHandler
    public void handle(@Payload String message, AmqpMessageHeaderAccessor messageHeaderAccessor, Channel channel) throws IOException {
        System.out.println(message);
        /**
         * 单条消息确认
         */
        try{
            /**
             * 确认消息
             * basicNack(long deliveryTag, boolean multiple, boolean requeue)
             * deliveryTag 接收标签
             * multiple 是否批量回滚
             */
            channel.basicAck(messageHeaderAccessor.getDeliveryTag(),false);
        }catch (Exception e){
            /**
             * 回滚确定
             * basicNack(long deliveryTag, boolean multiple, boolean requeue)
             * deliveryTag 接收标签
             * multiple 是否批量回滚
             * requeue true将消息回滚至队列，false：将消息设为死亡或移动至死信队列
             */
            new DirectExchange("");
            channel.basicNack(messageHeaderAccessor.getDeliveryTag(),false,true);
        }
    }

    @RabbitHandler
    public void handle(@Payload byte[] bytes,AmqpMessageHeaderAccessor messageHeaderAccessor, Channel channel) throws IOException {
        System.out.println(new String(bytes,"utf-8"));
        /**
         * 批量消息确认，将以前已经消费但未提交确认消息，全部确认
         */
        channel.basicAck(messageHeaderAccessor.getDeliveryTag(),true);
    }

}
